package com.yp.infrastructure.axon.config;


import com.mongodb.client.MongoClient;
import com.thoughtworks.xstream.XStream;
import com.yp.infrastructure.axon.interceptor.IAxonCommandHandlerInterceptor;
import com.yp.infrastructure.axon.interceptor.command.CmdMessageDispatcherInterceptor;
import com.yp.infrastructure.axon.interceptor.command.CmdMessageHandlerInterceptor;
import lombok.extern.slf4j.Slf4j;
import org.axonframework.commandhandling.CommandBus;
import org.axonframework.common.transaction.TransactionManager;
import org.axonframework.config.ConfigurerModule;
import org.axonframework.config.EventProcessingConfigurer;
import org.axonframework.disruptor.commandhandling.DisruptorCommandBus;
import org.axonframework.eventsourcing.eventstore.EmbeddedEventStore;
import org.axonframework.eventsourcing.eventstore.EventStorageEngine;
import org.axonframework.eventsourcing.eventstore.EventStore;
import org.axonframework.extensions.mongo.DefaultMongoTemplate;
import org.axonframework.extensions.mongo.eventsourcing.eventstore.MongoEventStorageEngine;
import org.axonframework.metrics.GlobalMetricRegistry;
import org.axonframework.serialization.xml.XStreamSerializer;
import org.axonframework.tracing.SpanFactory;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import javax.annotation.Resource;
import java.util.List;

@Slf4j
@Configuration
public class AxonClientConfiguration {


    @Resource
    private IAxonCommandHandlerInterceptor axonCommandHandlerInterceptor;

    /**
     * 配置事件处理器.
     * subscribingEventProcessors(SEP) 是订阅事件处理器，用于执行程序运行过程中产生的业务事件
     * trackingEventProcessors(TEP), 用于回放已经产生的事情，通常在程序启动时执行
     * 在系统启动阶段，通常会有大量的历史事件，业务上也不需要在程序启动时回放，故此配置未使用TEP.
     * TEP仅作用于EventHandler标签，若已使用了TEP,且不需要回放，在EventHandler标签上加上@DisallowedReplay.
     */

    @Bean
    public ConfigurerModule processorDefaultConfigurerModule() {
        log.debug("config SEP.");
        return configurer -> configurer.eventProcessing(EventProcessingConfigurer::usingSubscribingEventProcessors);
    }

    @Bean
    public EventStorageEngine storageEngine(MongoClient client) {
        log.debug("set up event store engine");
        XStream xStream = SecureXStreamSerializer.get().getXStream();
        XStreamSerializer serializer = XStreamSerializer.builder().xStream(xStream).build();
        return MongoEventStorageEngine.builder()
                .mongoTemplate(DefaultMongoTemplate.builder()
                        .mongoDatabase(client)
                        .build())
                .snapshotSerializer(serializer)
                .eventSerializer(serializer)
                .build();
    }

    /**
     * Axon provides an event store `EmbeddedEventStore`.
     * It delegates actual storage and retrieval of events to an `EventStorageEngine`.
     *
     * @return EmbeddedEventStore
     */
    @Bean
    public EventStore eventStore(EventStorageEngine storageEngine, GlobalMetricRegistry metricRegistry, SpanFactory spanFactory) {
        log.debug("set mongodb as event store.");
        return EmbeddedEventStore.builder()
                .storageEngine(storageEngine)
                .messageMonitor(metricRegistry.registerEventBus("mongoEventStore"))
                .spanFactory(spanFactory)
                .build();
    }


    @Bean
    public CommandBus commandBus(TransactionManager transactionManager,
                                 GlobalMetricRegistry metricRegistry) {
        log.debug("disruptor command bus");
        return DisruptorCommandBus.builder()
                .transactionManager(transactionManager)
                .invokerInterceptors(List.of(new CmdMessageHandlerInterceptor(axonCommandHandlerInterceptor)))
                .dispatchInterceptors(List.of(new CmdMessageDispatcherInterceptor(axonCommandHandlerInterceptor)))
                .messageMonitor(metricRegistry.registerCommandBus("disruptorCommandBus"))
                .bufferSize(4096)
                .build();
    }


}
